package com.unitd.scheduler.registry.impl;

import com.alibaba.fastjson.JSON;
import com.unitd.frame.comm.utils.StringUtils;
import com.unitd.scheduler.job.AbstractJob;
import com.unitd.scheduler.job.JobContext;
import com.unitd.scheduler.model.JobConfig;
import com.unitd.scheduler.monitor.MonitorCommond;
import com.unitd.scheduler.registry.IJobRegistry;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.zookeeper.data.Stat;
import org.quartz.CronExpression;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @desc 一致性服务采用zookeeper后,定时服务与zookeeper进行交互并处理定时任务执行控制的接口实现类
 * 此类中,主要用于控制定时服务与zookeeper的注册、销毁、服务的初始化、调用、销毁等处理
 * @filename ZkJobRegistry.java
 * @copyright www.unitd.com
 * @author Hudan
 * @version 1.0
 * @date 2016/10/10
 */
public class ZkJobRegistry implements InitializingBean, DisposableBean, IJobRegistry {

	private static final Logger logger = LoggerFactory.getLogger(ZkJobRegistry.class);

	/** 任务调度器的配置信息集合 */
	private Map<String, JobConfig> schedulerConfgs = new ConcurrentHashMap<>();
	/** 任务调度在一致性服务端(zookeeper/redis)存储的根目录 */
	public static final String ROOT = "/schedulers/";
	/** ookeeper服务器地址信息(IP,端口等) */
	private String zkServers;
	/** zookeeper客户端对象 */
	private ZkClient zkClient;
	/** 定时任务组所在的服务路径 */
	private String groupPath;
	/** 定时服务的节点状态文件所在的父目录 */
	private String nodeStateParentPath;
	/** zookeeper与定时服务的心跳检查服务 */
	private ScheduledExecutorService zkCheckTask;
	/** zookeeper服务是否可用 */
	private volatile boolean zkAvailabled = true;
	/** 是否正在服务状态更新中 */
	private volatile boolean updatingStatus;
	/** 节点时间是否已订阅 */
	private boolean nodeEventSubscribed = false;

	/**
	 * @desc 实现InitializingBean接口, 重写读取properties属性文件后的处理方法
	 */
	@Override
	public void afterPropertiesSet() throws Exception {
		ZkConnection zkConnection = new ZkConnection(zkServers);
		zkClient = new ZkClient(zkConnection, 10000);
//		zkCheckTask = Executors.newScheduledThreadPool(1);
		zkCheckTask = new ScheduledThreadPoolExecutor(1, new BasicThreadFactory.Builder().namingPattern("example-schedule-pool-%d").daemon(true).build());

		zkCheckTask.scheduleAtFixedRate(() -> {
			if (schedulerConfgs.isEmpty())
				return;
			List<String> activeNodes;
			try {
				activeNodes = zkClient.getChildren(nodeStateParentPath);
				zkAvailabled = true;
			} catch (Exception e) {
				// 检查zookeeper服务是否可用
				checkZkAvailabled();
				// 从可用的zookeeper服务中获取处于服务中的定时任务服务
				activeNodes = new ArrayList<>(JobContext.getContext().getActiveNodes());
			}

			if (!activeNodes.contains(JobContext.getContext().getNodeId())) {
				zkClient.createEphemeral(nodeStateParentPath + "/" + JobContext.getContext().getNodeId());
				logger.info("node[{}] re-join task clusters", JobContext.getContext().getNodeId());
			}

			// 对节点列表排序
			Collections.sort(activeNodes);
			// 本地缓存的所有jobs
			Collection<JobConfig> jobConfigs = schedulerConfgs.values();

			for (JobConfig jobConfig : jobConfigs) {
				// 如果本地任务指定的执行节点不在当前实际的节点列表，重新指定
				if (!activeNodes.contains(jobConfig.getCurrentNodeId())) {
					// 指定当前节点为排序后的第一个节点
					String newExecuteNodeId = activeNodes.get(0);
					jobConfig.setCurrentNodeId(newExecuteNodeId);
					logger.warn("Job[{}-{}] currentNodeId[{}] not in activeNodeList, assign new ExecuteNodeId:{}", jobConfig.getGroupName(), jobConfig.getJobName(), jobConfig.getCurrentNodeId(), newExecuteNodeId);
				}
			}
		}, 60, 30, TimeUnit.SECONDS);
	}

	/**
	 * @desc 继承Spring的DisposableBean类的destroy方法,每次调用完成后,都会去自动调用此方法
	 */
	@Override
	public void destroy() {
		// 关闭zookeeper与定时服务的心跳检查服务线程
		zkCheckTask.shutdown();
		// 关闭定时服务与zookeeper的连接
		zkClient.close();
	}

	/**
	 * @param conf 定时任务配置信息
	 * @desc 向一致性服务端(zookeeper/redis)注册定时服务
	 */
	@Override
	public void register(JobConfig conf) {
		// 是否第一个启动节点
		boolean isFirstNode = false;

		Calendar now = Calendar.getInstance();
		long currentTimeMillis = now.getTimeInMillis();
		conf.setModifyTime(currentTimeMillis);

		if (groupPath == null) {
			groupPath = ROOT + conf.getGroupName();
		}
		if (nodeStateParentPath == null) {
			nodeStateParentPath = groupPath + "/nodes";
		}

		String path = getPath(conf);

		final String jobName = conf.getJobName();

		if (!zkClient.exists(nodeStateParentPath)) {
			isFirstNode = true;
			zkClient.createPersistent(nodeStateParentPath, true);
		} else {
			// 检查是否有节点
			if (!isFirstNode) {
				isFirstNode = zkClient.getChildren(nodeStateParentPath).size() == 0;
			}
		}

		if (!zkClient.exists(path)) {
			zkClient.createPersistent(path, true);
		}

		// 是否要更新ZK的conf配置
		boolean updateConfInZK = isFirstNode;
		if (!updateConfInZK) {
			JobConfig configFromZK = getConfigFromZK(path, null);
			if (configFromZK != null) {
				/*
				 * 1.当前执行时间策略变化了
				 * 2.下一次执行时间在当前时间之前
				 * 3.配置文件修改是30分钟前
				 */
				if (!StringUtils.equals(configFromZK.getCronExpr(), conf.getCronExpr())) {
					updateConfInZK = true;
				} else if (configFromZK.getNextFireTime() != null && configFromZK.getNextFireTime().before(now.getTime())) {
					updateConfInZK = true;
				} else if (currentTimeMillis - configFromZK.getModifyTime() > TimeUnit.MINUTES.toMillis(30)) {
					updateConfInZK = true;
				} else {
					if (!JobContext.getContext().getNodeId().equals(configFromZK.getCurrentNodeId())) {
						List<String> nodes = zkClient.getChildren(nodeStateParentPath);
						updateConfInZK = !nodes.contains(configFromZK.getCurrentNodeId());
					}
				}
			} else {
				// zookeeper 该job不存在？
				updateConfInZK = true;
			}
			// 拿ZK上的配置覆盖当前的
			if (!updateConfInZK) {
				conf = configFromZK;
			}
		}

		if (updateConfInZK) {
			conf.setCurrentNodeId(JobContext.getContext().getNodeId());
			zkClient.writeData(path, JSON.toJSONString(conf));
		}
		schedulerConfgs.put(conf.getJobName(), conf);

		// 订阅同步信息变化
		zkClient.subscribeDataChanges(path, new IZkDataListener() {

			@Override
			public void handleDataDeleted(String dataPath) throws Exception {
				schedulerConfgs.remove(jobName);
			}

			@Override
			public void handleDataChange(String dataPath, Object data) throws Exception {
				if (data == null) return;
				JobConfig _jobConfig = JSON.parseObject(data.toString(), JobConfig.class);
				schedulerConfgs.put(jobName, _jobConfig);
			}
		});

		// 订阅节点事件
		regAndSubscribeNodeEvent();

		logger.info("finish register schConfig:{}", ToStringBuilder.reflectionToString(conf, ToStringStyle.MULTI_LINE_STYLE));
	}

	/**
	 * @desc 获取所有的定时任务配置信息
	 * @return 定时任务配置信息列表
	 */
	@Override
	public List<JobConfig> getAllJobs() {
		return new ArrayList<>(schedulerConfgs.values());
	}

	/**
	 * @desc 根据指定的定时任务名称获取该定时任务的配置信息
	 * 如果时强制从远程一致性服务端(zookeeper/redis)获取,则需要重新同步远程与本地服务的定时任务配置信息之后再返回
	 * @param jobName     定时任务名称
	 * @param forceRemote 是否强制从远程的一致性服务端(zookeeper/redis)获取和同步
	 * @return JobConfig
	 */
	@Override
	public JobConfig getConf(String jobName, boolean forceRemote) {
		JobConfig config = schedulerConfgs.get(jobName);

		if (forceRemote) {
			// 如果只有一个节点就不从强制同步了
			if (JobContext.getContext().getActiveNodes().size() == 1) {
				config.setCurrentNodeId(JobContext.getContext().getNodeId());
				return config;
			}
			String path = getPath(config);
			try {
				config = getConfigFromZK(path, null);
			} catch (Exception e) {
				checkZkAvailabled();
				logger.warn("fecth JobConfig from Registry error", e);
			}
		}
		return config;
	}

	/**
	 * @desc 更新定时任务的配置信息
	 * @param conf 定时任务配置信息
	 */
	@Override
	public void updateJobConfig(JobConfig conf) {
		conf.setModifyTime(Calendar.getInstance().getTimeInMillis());
		zkClient.writeData(getPath(conf), JSON.toJSONString(conf));
		schedulerConfgs.put(conf.getJobName(), conf);
	}

	/**
	 * @desc 设置需要运行的定时任务
	 * @param jobName  定时任务名称
	 * @param fireTime 首次启动时间
	 */
	@Override
	public void setRuning(String jobName, Date fireTime) {
		updatingStatus = false;
		try {
			JobConfig config = getConf(jobName, false);
			config.setRunning(true);
			config.setLastFireTime(fireTime);
			config.setModifyTime(Calendar.getInstance().getTimeInMillis());
			config.setErrorMsg(null);
			// 更新本地
			schedulerConfgs.put(jobName, config);
			try {
				if (zkAvailabled) zkClient.writeData(getPath(config), JSON.toJSONString(config));
			} catch (Exception e) {
				checkZkAvailabled();
				logger.warn(String.format("Job[{}] setRuning error...", jobName), e);
			}
		} finally {
			updatingStatus = false;
		}
	}

	/**
	 * @desc 停止定时任务
	 * @param jobName      定时任务名称
	 * @param nextFireTime 下次启动时间
	 * @param errorMsg     异常消息
	 */
	@Override
	public void setStoping(String jobName, Date nextFireTime, String errorMsg) {
		updatingStatus = false;
		try {
			JobConfig config = getConf(jobName, false);
			config.setRunning(false);
			config.setNextFireTime(nextFireTime);
			config.setModifyTime(Calendar.getInstance().getTimeInMillis());
			config.setErrorMsg(errorMsg);
			// 更新本地
			schedulerConfgs.put(jobName, config);
			try {
				if (zkAvailabled) zkClient.writeData(getPath(config), JSON.toJSONString(config));
			} catch (Exception ex) {
				checkZkAvailabled();
				logger.warn(String.format("Job[{}] setStoping error...", jobName), ex);
			}
		} finally {
			updatingStatus = false;
		}
	}

	/**
	 * @desc 从一致性服务端(zookeeper/redis)中将注册定时服务删除
	 * @param jobName 定时任务名称
	 */
	@Override
	public void unregister(String jobName) {
		JobConfig config = schedulerConfgs.get(jobName);

		String path = getPath(config);

		if (zkClient.getChildren(nodeStateParentPath).size() == 1) {
			zkClient.delete(path);
			logger.info("all node is closed ,delete path:" + path);
		}
	}

	/**
	 * @desc 从一致性服务端(zookeeper/redis)中将指定路径下的已注册的定时任务删除
	 */
	@Override
	public void onRegistered() {
		logger.info("==============clear Invalid jobs=================");
		List<String> jobs = zkClient.getChildren(groupPath);

		List<String> registerJobs = new ArrayList<>(JobContext.getContext().getAllJobs().keySet());
		String groupName = JobContext.getContext().getGroupName();
		String jobPath;
		for (String job : jobs) {
			if ("nodes".equals(job)) continue;

			if (registerJobs.contains(groupName + ":" + job)) continue;
			jobPath = groupPath + "/" + job;
			zkClient.delete(jobPath);
		}
		logger.info("==============clear Invalid jobs end=================");
	}

	public void setZkServers(String zkServers) {
		this.zkServers = zkServers;
	}

	/**
	 * @desc 检查zookeeper服务是否可用
	 * @return boolean
	 */
	private boolean checkZkAvailabled() {
		try {
			zkClient.exists(ROOT);
			zkAvailabled = true;
		} catch (Exception e) {
			zkAvailabled = false;
			logger.warn("ZK server is not available....");
		}
		return zkAvailabled;
	}

	/**
	 * @desc 获取定时服务中定时任务的配置信息在zookeeper中的详细路径
	 * @param config 定时任务配置西南西
	 * @return 配置信息在zookeeper中的详细地址
	 */
	private String getPath(JobConfig config) {
		return ROOT + config.getGroupName() + "/" + config.getJobName();
	}

	/**
	 * @desc 根据定时任务的配置信息在zookeeper中的详细地址和状态,从zookeeper中获取该定时任务的配置信息
	 * @param path 配置信息在zookeeper中的详细地址
	 * @param stat 状态信息
	 * @return 定时任务的配置信息
	 */
	private synchronized JobConfig getConfigFromZK(String path, Stat stat) {
		Object data = stat == null ? zkClient.readData(path) : zkClient.readData(path, stat);
		return data == null ? null : JSON.parseObject(data.toString(), JobConfig.class);
	}

	/**
	 * @desc 订阅节点事件
	 */
	private synchronized void regAndSubscribeNodeEvent() {
		if (nodeEventSubscribed) return;
		// 创建node节点
		zkClient.createEphemeral(nodeStateParentPath + "/" + JobContext.getContext().getNodeId());

		String path;
		// 订阅节点信息变化
		zkClient.subscribeChildChanges(nodeStateParentPath, new IZkChildListener() {
			@Override
			public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
				if (currentChilds == null || !currentChilds.contains(JobContext.getContext().getNodeId())) {
					zkClient.createEphemeral(nodeStateParentPath + "/" + JobContext.getContext().getNodeId());
					logger.info("Nodelist is empty~ node[{}] re-join task clusters", JobContext.getContext().getNodeId());
					return;
				}
				logger.info(">>nodes changed ,nodes:{}", currentChilds);

				// 分配节点
				rebalanceJobNode(currentChilds);
				// 刷新当前可用节点
				JobContext.getContext().refreshNodes(currentChilds);
			}
		});

		logger.info("subscribe nodes change event at path:{}", nodeStateParentPath);

		// 注册手动执行事件监听(来自于监控平台的)
		path = nodeStateParentPath + "/" + JobContext.getContext().getNodeId();
		zkClient.subscribeDataChanges(path, new IZkDataListener() {
			@Override
			public void handleDataDeleted(String dataPath) throws Exception {
			}

			@Override
			public void handleDataChange(String dataPath, Object data) throws Exception {
				MonitorCommond cmd = (MonitorCommond) data;
				if (cmd != null) {
					logger.info("收到commond:" + cmd.toString());
					execCommond(cmd);
				}
			}
		});
		logger.info("subscribe command event at path:{}", path);

		// 刷新节点列表
		List<String> activeNodes = zkClient.getChildren(nodeStateParentPath);
		JobContext.getContext().refreshNodes(activeNodes);

		logger.info("current activeNodes:{}", activeNodes);
		nodeEventSubscribed = true;
	}

	/**
	 * @desc 重新分配执行节点
	 * @param nodes 可用的定时服务节点
	 */
	private synchronized void rebalanceJobNode(List<String> nodes) {
		while (updatingStatus) ;
		Collection<JobConfig> jobs = schedulerConfgs.values();
		int nodeIndex = 0;
		for (JobConfig job : jobs) {
			String nodeId = nodes.get(nodeIndex++);
			if (!StringUtils.equals(job.getCurrentNodeId(), nodeId)) {
				job.setCurrentNodeId(nodeId);
				logger.info("rebalance Job[{}-{}] To Node[{}] ", job.getGroupName(), job.getJobName(), nodeId);
			}
			if (nodeIndex >= nodes.size()) {
				nodeIndex = 0;
			}
			updateJobConfig(job);
		}
	}

	/**
	 * @desc 执行响应的定时任务的命令
	 * @param cmd 命令
	 */
	private void execCommond(MonitorCommond cmd) {
		if (cmd == null) return;

		JobConfig config = schedulerConfgs.get(cmd.getJobName());
		String key = cmd.getJobGroup() + ":" + cmd.getJobName();
		final AbstractJob abstractJob = JobContext.getContext().getAllJobs().get(key);
		if (MonitorCommond.TYPE_EXEC == cmd.getCmdType()) {
			if (config.isRunning()) {
				logger.info("任务正在执行中,请稍后再执行");
				return;
			}
			if (abstractJob != null) {
				JobContext.getContext().submitSyncTask(new Runnable() {
					@Override
					public void run() {
						try {
							logger.info("begin execute job[{}] by MonitorCommond", abstractJob.getJobName());
							abstractJob.doJob(JobContext.getContext());
						} catch (Exception e) {
							logger.error(abstractJob.getJobName(), e);
						}
					}
				});
			} else {
				logger.warn("Not found job by key:{} !!!!", key);
			}
		} else if (MonitorCommond.TYPE_STATUS_MOD == cmd.getCmdType()
				|| MonitorCommond.TYPE_CRON_MOD == cmd.getCmdType()) {

			if (config != null) {
				if (MonitorCommond.TYPE_STATUS_MOD == cmd.getCmdType()) {
					config.setActive("1".equals(cmd.getBody()));
				} else {
					try {
						new CronExpression(cmd.getBody().toString());
					} catch (Exception e) {
						throw new RuntimeException("cron表达式格式错误");
					}
					abstractJob.resetTriggerCronExpr(cmd.getBody().toString());
					config.setCronExpr(cmd.getBody().toString());

				}
				updateJobConfig(config);
				if (JobContext.getContext().getConfigPersistHandler() != null) {
					JobContext.getContext().getConfigPersistHandler().persist(config);
				}
			}
		}
	}
}